diff --git a/components/components.go b/components/components.go
index 6b784ad5..06eb0720 100644
--- a/components/components.go
+++ b/components/components.go
@@ -117,6 +117,7 @@ import (
 	"github.com/SigNoz/signoz-otel-collector/exporter/clickhousetracesexporter"
 	"github.com/SigNoz/signoz-otel-collector/exporter/signozkafkaexporter"
 	_ "github.com/SigNoz/signoz-otel-collector/pkg/parser/grok"
+	"github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor"
 	"github.com/SigNoz/signoz-otel-collector/processor/signozspanmetricsprocessor"
 	"github.com/SigNoz/signoz-otel-collector/processor/signoztailsampler"
 	"github.com/SigNoz/signoz-otel-collector/processor/signoztransformprocessor"
@@ -255,6 +256,7 @@ func Components() (otelcol.Factories, error) {
 		logstransformprocessor.NewFactory(),
 		signoztailsampler.NewFactory(),
 		signoztransformprocessor.NewFactory(),
+		signozlogspipelineprocessor.NewFactory(),
 	}
 	for _, pr := range factories.Processors {
 		processors = append(processors, pr)
diff --git a/processor/signozlogspipelineprocessor/README.md b/processor/signozlogspipelineprocessor/README.md
new file mode 100644
index 00000000..6ab68bd7
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/README.md
@@ -0,0 +1,20 @@
+# Signoz Logs Pipeline Processor
+
+An otel processor for powering SigNoz logs pipelines.
+
+The goal of this processor is to achieve a seamless logs pipelines experience for SigNoz users
+while exposing the same interface as the logstransform processor in opentelemetry-collector-contrib
+
+The implementation mostly borrows from, adapts and enhances the official logstransform processor (and stanza)
+code to achieve goals of SigNoz pipelines
+
+On success, query service should be able to add logs pipelines to otel collector config using
+this processor without having to do much translation or augmentation of the user defined config
+
+The following non-exhaustive list of capabilities/improvements on top of what logstransform provides are included
+- sync ConsumeLogs i.e. ConsumeLogs should only return after the ConsumeLogs of the next processor in the pipeline has returned.
+  This helps with reliability of logs processing through the collector.
+  The otel-collector-contrib logstransform processor doesn't have a sync implementation for ConsumeLogs
+- Enhanced addressing capabilities.
+  - should be able to refer to top level otlp log fields (like severity) in operator conditions
+  - should be able to use body.field in operator config if body is JSON without having to first parse body into an temporary attribute
\ No newline at end of file
diff --git a/processor/signozlogspipelineprocessor/config.go b/processor/signozlogspipelineprocessor/config.go
new file mode 100644
index 00000000..d7819703
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/config.go
@@ -0,0 +1,23 @@
+// Brought in as is from logstransform processor in opentelemetry-collector-contrib
+package signozlogspipelineprocessor
+
+import (
+	"errors"
+
+	"go.opentelemetry.io/collector/component"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
+)
+
+type Config struct {
+	adapter.BaseConfig `mapstructure:",squash"`
+}
+
+var _ component.Config = (*Config)(nil)
+
+func (cfg *Config) Validate() error {
+	if len(cfg.BaseConfig.Operators) == 0 {
+		return errors.New("no operators were configured for signozlogspipeline processor")
+	}
+	return nil
+}
diff --git a/processor/signozlogspipelineprocessor/config_test.go b/processor/signozlogspipelineprocessor/config_test.go
new file mode 100644
index 00000000..29c6cda6
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/config_test.go
@@ -0,0 +1,48 @@
+// Brought in as is from logstransform processor in opentelemetry-collector-contrib
+package signozlogspipelineprocessor
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/parser/regex"
+)
+
+func TestLoadConfig(t *testing.T) {
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
+	require.NoError(t, err)
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.NoError(t, component.UnmarshalConfig(cm, cfg))
+	assert.Equal(t, &Config{
+		BaseConfig: adapter.BaseConfig{
+			Operators: []operator.Config{
+				{
+					Builder: func() *regex.Config {
+						cfg := regex.NewConfig()
+						cfg.Regex = "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"
+						sevField := entry.NewAttributeField("sev")
+						sevCfg := helper.NewSeverityConfig()
+						sevCfg.ParseFrom = &sevField
+						cfg.SeverityConfig = &sevCfg
+						timeField := entry.NewAttributeField("time")
+						timeCfg := helper.NewTimeParser()
+						timeCfg.Layout = "%Y-%m-%d %H:%M:%S"
+						timeCfg.ParseFrom = &timeField
+						cfg.TimeParser = &timeCfg
+						return cfg
+					}(),
+				},
+			},
+		},
+	}, cfg)
+}
diff --git a/processor/signozlogspipelineprocessor/factory.go b/processor/signozlogspipelineprocessor/factory.go
new file mode 100644
index 00000000..f6fce3ee
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/factory.go
@@ -0,0 +1,48 @@
+// Brought in as is from logstransform processor in opentelemetry-collector-contrib
+// with identifiers changed for the new processor
+package signozlogspipelineprocessor
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+)
+
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		component.MustNewType("signozlogspipeline"),
+		createDefaultConfig,
+		processor.WithLogs(createLogsProcessor, component.StabilityLevelDevelopment))
+}
+
+// Note: This isn't a valid configuration (no operators would lead to no work being done)
+func createDefaultConfig() component.Config {
+	return &Config{
+		BaseConfig: adapter.BaseConfig{
+			Operators: []operator.Config{},
+		},
+	}
+}
+
+func createLogsProcessor(
+	_ context.Context,
+	set processor.CreateSettings,
+	cfg component.Config,
+	nextConsumer consumer.Logs) (processor.Logs, error) {
+	pCfg, ok := cfg.(*Config)
+	if !ok {
+		return nil, errors.New("could not initialize signozlogspipeline processor")
+	}
+
+	if len(pCfg.BaseConfig.Operators) == 0 {
+		return nil, errors.New("no operators were configured for signozlogspipeline processor")
+	}
+
+	return newProcessor(pCfg, nextConsumer, set.TelemetrySettings)
+}
diff --git a/processor/signozlogspipelineprocessor/factory_test.go b/processor/signozlogspipelineprocessor/factory_test.go
new file mode 100644
index 00000000..4e62ea0b
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/factory_test.go
@@ -0,0 +1,72 @@
+// Brought in as is from logstransform processor in opentelemetry-collector-contrib
+package signozlogspipelineprocessor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/processor/processortest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/parser/regex"
+)
+
+func TestCreateDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.NoError(t, componenttest.CheckConfigStruct(cfg))
+	assert.NotNil(t, cfg)
+}
+
+func TestCreateProcessor(t *testing.T) {
+	factory := NewFactory()
+	cfg := &Config{
+		BaseConfig: adapter.BaseConfig{
+			Operators: []operator.Config{
+				{
+					Builder: func() *regex.Config {
+						cfg := regex.NewConfig()
+						cfg.Regex = "^(?P<time>\\d{4}-\\d{2}-\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"
+						sevField := entry.NewAttributeField("sev")
+						sevCfg := helper.NewSeverityConfig()
+						sevCfg.ParseFrom = &sevField
+						cfg.SeverityConfig = &sevCfg
+						timeField := entry.NewAttributeField("time")
+						timeCfg := helper.NewTimeParser()
+						timeCfg.Layout = "%Y-%m-%d"
+						timeCfg.ParseFrom = &timeField
+						cfg.TimeParser = &timeCfg
+						return cfg
+					}(),
+				},
+			},
+		},
+	}
+
+	tp, err := factory.CreateLogsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg, consumertest.NewNop())
+	assert.NoError(t, err)
+	assert.NotNil(t, tp)
+}
+
+func TestInvalidOperators(t *testing.T) {
+	factory := NewFactory()
+	cfg := &Config{
+		BaseConfig: adapter.BaseConfig{
+			Operators: []operator.Config{
+				{
+					// invalid due to missing regex
+					Builder: regex.NewConfig(),
+				},
+			},
+		},
+	}
+
+	_, err := factory.CreateLogsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg, nil)
+	assert.Error(t, err)
+}
diff --git a/processor/signozlogspipelineprocessor/processor.go b/processor/signozlogspipelineprocessor/processor.go
new file mode 100644
index 00000000..8f289425
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/processor.go
@@ -0,0 +1,209 @@
+// A sync implementation of stanza based otel collector processor.Logs
+// logstransform processor in opentelemetry-collector-contrib is async
+package signozlogspipelineprocessor
+
+import (
+	"context"
+	"encoding/binary"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/pipeline"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/extension/experimental/storage"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+
+	_ "github.com/SigNoz/signoz-otel-collector/pkg/parser/grok" // ensure grok parser gets registered.
+)
+
+func newProcessor(
+	processorConfig *Config,
+	nextConsumer consumer.Logs,
+	telemetrySettings component.TelemetrySettings,
+) (*logsPipelineProcessor, error) {
+
+	sink := &stanzaToOtelConsumer{
+		nextConsumer: nextConsumer,
+		logger:       telemetrySettings.Logger,
+	}
+
+	stanzaPipeline, err := pipeline.Config{
+		Operators:     processorConfig.BaseConfig.Operators,
+		DefaultOutput: sink,
+	}.Build(telemetrySettings)
+	if err != nil {
+		return nil, err
+	}
+
+	return &logsPipelineProcessor{
+		telemetrySettings: telemetrySettings,
+		nextConsumer:      nextConsumer,
+
+		processorConfig: processorConfig,
+		stanzaPipeline:  stanzaPipeline,
+
+		sink: sink,
+	}, nil
+}
+
+// Implements processor.Logs
+type logsPipelineProcessor struct {
+	telemetrySettings component.TelemetrySettings
+	nextConsumer      consumer.Logs
+
+	processorConfig *Config
+	stanzaPipeline  *pipeline.DirectedPipeline
+	firstOp         operator.Operator
+
+	shutdownFns []component.ShutdownFunc
+
+	sink *stanzaToOtelConsumer
+}
+
+// Collector starting up
+func (p *logsPipelineProcessor) Start(ctx context.Context, _ component.Host) error {
+	p.telemetrySettings.Logger.Info("Starting logsPipelineProcessor")
+
+	err := p.stanzaPipeline.Start(storage.NewNopClient())
+	if err != nil {
+		return err
+	}
+
+	p.shutdownFns = append(p.shutdownFns, func(_ context.Context) error {
+		return p.stanzaPipeline.Stop()
+	})
+
+	// .Operators() returns topologically sorted operators for the stanza operator graph
+	// logs will be fed into p.firstOp for processing
+	p.firstOp = p.stanzaPipeline.Operators()[0]
+
+	return nil
+}
+
+// Collector shutting down
+func (p *logsPipelineProcessor) Shutdown(ctx context.Context) error {
+	p.telemetrySettings.Logger.Info("Stopping logsPipelineProcessor")
+
+	// Call shutdown functions in reverse order, so that shutdown happens in LIFO order of starting
+	for i := len(p.shutdownFns) - 1; i >= 0; i-- {
+		fn := p.shutdownFns[i]
+
+		if err := fn(ctx); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// baseConsumer interface
+func (p *logsPipelineProcessor) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: true}
+}
+
+func (p *logsPipelineProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
+	p.telemetrySettings.Logger.Debug(
+		"logsPipelineProcessor received logs",
+		zap.Any("logs", ld),
+	)
+
+	entries := plogToEntries(ld)
+
+	for _, e := range entries {
+		if err := p.firstOp.Process(ctx, e); err != nil {
+			p.telemetrySettings.Logger.Error("processor encountered an issue with the pipeline", zap.Error(err))
+		}
+	}
+
+	// Flush processed entries as a single plog.Logs to next consumer.
+	return p.sink.flush(ctx)
+}
+
+// Helpers below have been brought in as is from stanza adapter for logstransform
+func plogToEntries(src plog.Logs) []*entry.Entry {
+	result := []*entry.Entry{}
+	for rlIdx := 0; rlIdx < src.ResourceLogs().Len(); rlIdx++ {
+		resourceLogs := src.ResourceLogs().At(rlIdx)
+		resourceAttribs := resourceLogs.Resource().Attributes().AsRaw()
+
+		for slIdx := 0; slIdx < resourceLogs.ScopeLogs().Len(); slIdx++ {
+			scopeLogs := resourceLogs.ScopeLogs().At(slIdx)
+
+			for lrIdx := 0; lrIdx < scopeLogs.LogRecords().Len(); lrIdx++ {
+				record := scopeLogs.LogRecords().At(lrIdx)
+				entry := entry.Entry{}
+				entry.ScopeName = scopeLogs.Scope().Name()
+				entry.Resource = resourceAttribs
+				convertFrom(record, &entry)
+				result = append(result, &entry)
+
+			}
+		}
+	}
+
+	return result
+}
+
+func convertFrom(src plog.LogRecord, ent *entry.Entry) {
+	// if src.Timestamp == 0, then leave ent.Timestamp as nil
+	if src.Timestamp() != 0 {
+		ent.Timestamp = src.Timestamp().AsTime()
+	}
+
+	if src.ObservedTimestamp() == 0 {
+		ent.ObservedTimestamp = time.Now()
+	} else {
+		ent.ObservedTimestamp = src.ObservedTimestamp().AsTime()
+	}
+
+	ent.Severity = fromPdataSevMap[src.SeverityNumber()]
+	ent.SeverityText = src.SeverityText()
+
+	ent.Attributes = src.Attributes().AsRaw()
+	ent.Body = src.Body().AsRaw()
+
+	if !src.TraceID().IsEmpty() {
+		buffer := src.TraceID()
+		ent.TraceID = buffer[:]
+	}
+	if !src.SpanID().IsEmpty() {
+		buffer := src.SpanID()
+		ent.SpanID = buffer[:]
+	}
+	if src.Flags() != 0 {
+		a := make([]byte, 4)
+		binary.LittleEndian.PutUint32(a, uint32(src.Flags()))
+		ent.TraceFlags = []byte{a[0]}
+	}
+}
+
+var fromPdataSevMap = map[plog.SeverityNumber]entry.Severity{
+	plog.SeverityNumberUnspecified: entry.Default,
+	plog.SeverityNumberTrace:       entry.Trace,
+	plog.SeverityNumberTrace2:      entry.Trace2,
+	plog.SeverityNumberTrace3:      entry.Trace3,
+	plog.SeverityNumberTrace4:      entry.Trace4,
+	plog.SeverityNumberDebug:       entry.Debug,
+	plog.SeverityNumberDebug2:      entry.Debug2,
+	plog.SeverityNumberDebug3:      entry.Debug3,
+	plog.SeverityNumberDebug4:      entry.Debug4,
+	plog.SeverityNumberInfo:        entry.Info,
+	plog.SeverityNumberInfo2:       entry.Info2,
+	plog.SeverityNumberInfo3:       entry.Info3,
+	plog.SeverityNumberInfo4:       entry.Info4,
+	plog.SeverityNumberWarn:        entry.Warn,
+	plog.SeverityNumberWarn2:       entry.Warn2,
+	plog.SeverityNumberWarn3:       entry.Warn3,
+	plog.SeverityNumberWarn4:       entry.Warn4,
+	plog.SeverityNumberError:       entry.Error,
+	plog.SeverityNumberError2:      entry.Error2,
+	plog.SeverityNumberError3:      entry.Error3,
+	plog.SeverityNumberError4:      entry.Error4,
+	plog.SeverityNumberFatal:       entry.Fatal,
+	plog.SeverityNumberFatal2:      entry.Fatal2,
+	plog.SeverityNumberFatal3:      entry.Fatal3,
+	plog.SeverityNumberFatal4:      entry.Fatal4,
+}
diff --git a/processor/signozlogspipelineprocessor/processor_test.go b/processor/signozlogspipelineprocessor/processor_test.go
new file mode 100644
index 00000000..51738abc
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/processor_test.go
@@ -0,0 +1,392 @@
+package signozlogspipelineprocessor
+
+import (
+	"context"
+	"encoding/hex"
+	"testing"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/processor/processortest"
+	"gopkg.in/yaml.v2"
+)
+
+// Tests for processors (stanza operators) supported in log pipelines
+
+func TestAddProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: add
+      field: attributes.test
+      value: testValue
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		"test log", map[string]any{"test": "testValue"},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestRemoveProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: remove
+      field: attributes.test
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{"test": "testValue"},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		"test log", map[string]any{},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestMoveProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: move
+      from: attributes.test
+      to: attributes.test1
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{"test": "testValue"},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		"test log", map[string]any{"test1": "testValue"},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestCopyProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: copy
+      from: attributes.test
+      to: attributes.test1
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{"test": "testValue"},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		"test log", map[string]any{"test": "testValue", "test1": "testValue"},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestRegexProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: regex_parser
+      regex: ^a=(?P<a>.+);b=(?P<b>.+)$
+      parse_from: body
+      parse_to: attributes
+  `
+
+	input := []plog.Logs{makePlog(
+		"a=aval;b=bval", map[string]any{},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		"a=aval;b=bval", map[string]any{
+			"a": "aval",
+			"b": "bval",
+		},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestGrokProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: grok_parser
+      pattern: 'status: %{INT:status_code:int}'
+      parse_from: body
+      parse_to: attributes
+  `
+
+	input := []plog.Logs{makePlog(
+		"status: 200", map[string]any{},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		"status: 200", map[string]any{"status_code": 200},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestJSONProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: json_parser
+      parse_from: body
+      parse_to: attributes
+  `
+
+	input := []plog.Logs{makePlog(
+		`{"status": "ok"}`, map[string]any{},
+	)}
+	expectedOutput := []plog.Logs{makePlog(
+		`{"status": "ok"}`, map[string]any{"status": "ok"},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestTraceProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: trace_parser
+      trace_id:
+        parse_from: attributes.traceId
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{"traceId": "e37e734349000e2eda9c07cca0ceb692"},
+	)}
+	expectedOutput := []plog.Logs{makePlogWithTopLevelFields(
+		t, "test log", map[string]any{"traceId": "e37e734349000e2eda9c07cca0ceb692"},
+		map[string]any{
+			"trace_id": "e37e734349000e2eda9c07cca0ceb692",
+		},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestSeverityProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: severity_parser
+      parse_from: attributes.sev
+      mapping:
+        error: oops
+      overwrite_text: true
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{"sev": "oops"},
+	)}
+	expectedOutput := []plog.Logs{makePlogWithTopLevelFields(
+		t, "test log", map[string]any{"sev": "oops"},
+		map[string]any{
+			"severity_text":   "ERROR",
+			"severity_number": 17,
+		},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func TestTimeProcessor(t *testing.T) {
+	confYaml := `
+  operators:
+    - type: time_parser
+      parse_from: attributes.tsUnixEpoch
+      layout_type: epoch
+      layout: s
+      overwrite_text: true
+  `
+
+	input := []plog.Logs{makePlog(
+		"test log", map[string]any{"tsUnixEpoch": 9999},
+	)}
+	expectedOutput := []plog.Logs{makePlogWithTopLevelFields(
+		t, "test log", map[string]any{"tsUnixEpoch": 9999},
+		map[string]any{"timestamp": time.Unix(9999, 0)},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+// Test that translated Signoz logs pipelines work as expected
+func TestSignozLogPipelineWithRouterOp(t *testing.T) {
+	confYaml := `
+  operators:
+      - default: noop
+        id: router_signoz
+        routes:
+          - expr: attributes["container_name"] == "hotrod"
+            output: 61357da4-6d61-47f7-bcc5-1a2f00801b9a
+        type: router
+      - id: 61357da4-6d61-47f7-bcc5-1a2f00801b9a
+        if: body != nil && body matches "^(?P<ts>.*)\\t(?P<log_level>.*)\\t(?P<location>.*)\\t(?P<msg>.*)\\t(?P<data_json>.*)$$"
+        on_error: send
+        output: f2ff87ad-f051-46d4-89c3-0cef056d8677
+        parse_from: body
+        parse_to: attributes
+        regex: ^(?P<ts>.*)\t(?P<log_level>.*)\t(?P<location>.*)\t(?P<msg>.*)\t(?P<data_json>.*)$$
+        type: regex_parser
+      - id: f2ff87ad-f051-46d4-89c3-0cef056d8677
+        if: attributes?.data_json != nil && attributes.data_json matches "^\\s*{.*}\\s*$$"
+        output: dde26fcc-9140-4415-b1fb-d6daeab5e744
+        parse_from: attributes.data_json
+        parse_to: attributes
+        type: json_parser
+      - id: dde26fcc-9140-4415-b1fb-d6daeab5e744
+        if: attributes?.log_level != nil && ( type(attributes.log_level) == "string" || ( type(attributes.log_level) in ["int", "float"] && attributes.log_level == float(int(attributes.log_level)) ) )
+        mapping:
+          debug:
+              - DEBUG
+          error:
+              - ERROR
+          fatal:
+              - FATAL
+          info:
+              - INFO
+          trace:
+              - TRACE
+          warn:
+              - WARN
+        overwrite_text: true
+        parse_from: attributes.log_level
+        type: severity_parser
+      - id: 01ee2767-5d86-43c7-b0c4-a66651850b51
+        type: remove
+        if: attributes?.data_json != nil
+        field: attributes.data_json
+      - id: noop
+        type: noop
+  `
+
+	// A matching log should get processed.
+	input := []plog.Logs{makePlog(
+		`2024-09-04T09:58:39.635Z	ERROR	driver/server.go:85	Retrying GetDriver after error	{"service": "driver", "trace_id": "738d1c34020ba19e", "span_id": "69e77f208cb24e9b", "retry_no": 1, "error": "redis timeout"}`,
+		map[string]any{"container_name": "hotrod"},
+	)}
+	expectedOutput := []plog.Logs{makePlogWithTopLevelFields(
+		t,
+		`2024-09-04T09:58:39.635Z	ERROR	driver/server.go:85	Retrying GetDriver after error	{"service": "driver", "trace_id": "738d1c34020ba19e", "span_id": "69e77f208cb24e9b", "retry_no": 1, "error": "redis timeout"}`,
+		map[string]any{
+			"container_name": "hotrod",
+			"error":          "redis timeout",
+			"location":       "driver/server.go:85",
+			"log_level":      "ERROR",
+			"msg":            "Retrying GetDriver after error",
+			"retry_no":       float64(1),
+			"service":        "driver",
+			"span_id":        "69e77f208cb24e9b",
+			"trace_id":       "738d1c34020ba19e",
+			"ts":             "2024-09-04T09:58:39.635Z",
+		},
+		map[string]any{
+			"severity_text":   "ERROR",
+			"severity_number": 17,
+		},
+	)}
+
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+
+	// A non-matching log should get ignored.
+	input = []plog.Logs{makePlog(
+		`2024-09-04T09:58:39.635Z	ERROR	driver/server.go:85	Retrying GetDriver after error	{"service": "driver", "trace_id": "738d1c34020ba19e", "span_id": "69e77f208cb24e9b", "retry_no": 1, "error": "redis timeout"}`,
+		map[string]any{"container_name": "not-hotrod"},
+	)}
+	expectedOutput = []plog.Logs{makePlog(
+		`2024-09-04T09:58:39.635Z	ERROR	driver/server.go:85	Retrying GetDriver after error	{"service": "driver", "trace_id": "738d1c34020ba19e", "span_id": "69e77f208cb24e9b", "retry_no": 1, "error": "redis timeout"}`,
+		map[string]any{"container_name": "not-hotrod"},
+	)}
+	validateProcessorBehavior(t, confYaml, input, expectedOutput)
+}
+
+func validateProcessorBehavior(
+	t *testing.T,
+	confYaml string,
+	inputLogs []plog.Logs,
+	expectedOutput []plog.Logs,
+) {
+	require := require.New(t)
+
+	factory := NewFactory()
+
+	config := parseProcessorConfig(t, confYaml)
+	testSink := new(consumertest.LogsSink)
+	proc, err := factory.CreateLogsProcessor(
+		context.Background(),
+		processortest.NewNopCreateSettings(),
+		config, testSink,
+	)
+	require.NoError(err)
+
+	err = proc.Start(context.Background(), nil)
+	require.NoError(err)
+
+	for _, inputPlog := range inputLogs {
+		err = proc.ConsumeLogs(context.Background(), inputPlog)
+		require.NoError(err)
+	}
+
+	output := testSink.AllLogs()
+	require.Len(output, len(expectedOutput))
+	for i, expected := range expectedOutput {
+		found := output[i]
+		require.NoError(plogtest.CompareLogs(expected, found))
+	}
+}
+
+func parseProcessorConfig(t *testing.T, confYaml string) component.Config {
+	var rawConf map[string]any
+	err := yaml.Unmarshal([]byte(confYaml), &rawConf)
+	require.NoError(t, err, "couldn't parse config yaml")
+	cm := confmap.NewFromStringMap(rawConf)
+
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	require.NoError(
+		t, component.UnmarshalConfig(cm, cfg),
+		"couldn't unmarshal parsed yaml into processor config",
+	)
+	return cfg
+}
+
+func makePlog(body string, attributes map[string]any) plog.Logs {
+	ld := plog.NewLogs()
+	lr := ld.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+	lr.Body().SetStr(body)
+	lr.Attributes().FromRaw(attributes)
+
+	lr.SetObservedTimestamp(pcommon.NewTimestampFromTime(time.Unix(500, 0)))
+
+	return ld
+}
+
+func makePlogWithTopLevelFields(t *testing.T, body string, attributes map[string]any, fields map[string]any) plog.Logs {
+	ld := makePlog(body, attributes)
+	lr := ld.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0)
+
+	if traceId, exists := fields["trace_id"]; exists {
+		traceIdBytes, err := hex.DecodeString(traceId.(string))
+		require.NoError(t, err)
+
+		lr.SetTraceID(pcommon.TraceID(traceIdBytes))
+	}
+
+	if sevText, exists := fields["severity_text"]; exists {
+		lr.SetSeverityText(sevText.(string))
+	}
+	if sevNum, exists := fields["severity_number"]; exists {
+		lr.SetSeverityNumber(plog.SeverityNumber(sevNum.(int)))
+	}
+
+	if timestamp, exists := fields["timestamp"]; exists {
+		lr.SetTimestamp(pcommon.NewTimestampFromTime(timestamp.(time.Time)))
+	}
+
+	return ld
+}
diff --git a/processor/signozlogspipelineprocessor/stanza_consumer.go b/processor/signozlogspipelineprocessor/stanza_consumer.go
new file mode 100644
index 00000000..aa3762ad
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/stanza_consumer.go
@@ -0,0 +1,93 @@
+package signozlogspipelineprocessor
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"go.opentelemetry.io/collector/consumer"
+	"go.uber.org/zap"
+)
+
+// A stanza operator that consumes stanza entries and converts them to pdata.Log
+// for passing them on to the next otel consumer.
+type stanzaToOtelConsumer struct {
+	nextConsumer consumer.Logs
+	logger       *zap.Logger
+
+	// One plog.Logs can contain many log records. While the otel processor ConsumeLogs works
+	// with one plog.Logs at a time, the stanza pipeline works with one log entry at a time.
+	// `processedEntries` accumulates processed entries for a single plog.Logs until the
+	// signozlogspipeline processor flushes these entries out - converting them back into
+	// a single plog.Logs that gets sent to `nextConsumer.ConsumeLogs`
+	processedEntries []*entry.Entry
+}
+
+// stanzaToOtelConsumer must implement the stanza operator.Operator interface
+func (c *stanzaToOtelConsumer) ID() string {
+	return "stanza-otel-consumer"
+}
+
+func (c *stanzaToOtelConsumer) Type() string {
+	return "stanza-otel-consumer"
+}
+
+func (c *stanzaToOtelConsumer) Start(_ operator.Persister) error {
+	return nil
+}
+
+func (c *stanzaToOtelConsumer) Stop() error {
+	return nil
+}
+
+func (c *stanzaToOtelConsumer) CanOutput() bool {
+	return false
+}
+
+func (c *stanzaToOtelConsumer) Outputs() []operator.Operator {
+	return []operator.Operator{}
+}
+
+func (c *stanzaToOtelConsumer) GetOutputIDs() []string {
+	return []string{}
+}
+
+func (c *stanzaToOtelConsumer) SetOutputs([]operator.Operator) error {
+	return fmt.Errorf("outputs not supported")
+}
+
+func (c *stanzaToOtelConsumer) SetOutputIDs([]string) {
+}
+
+func (c *stanzaToOtelConsumer) CanProcess() bool {
+	return true
+}
+
+// Process an entry passed on to this operator by a stanza pipeline
+// This operator is expected to be used as a sink in the stanza pipeline
+func (c *stanzaToOtelConsumer) Process(ctx context.Context, entry *entry.Entry) error {
+	c.processedEntries = append(c.processedEntries, entry)
+	return nil
+}
+
+// One plog.Logs can contain many log records. While the otel processor ConsumeLogs works
+// with one plog.Logs at a time, the stanza pipeline works with one log entry at a time.
+// `processedEntries` accumulates processed entries for a single plog.Logs until the
+// signozlogspipeline processor flushes these entries out - converting them back into
+// a single plog.Logs that gets sent to `nextConsumer.ConsumeLogs`
+func (c *stanzaToOtelConsumer) flush(ctx context.Context) error {
+	plogs := convertEntriesToPlogs(c.processedEntries)
+	c.processedEntries = []*entry.Entry{}
+
+	err := c.nextConsumer.ConsumeLogs(ctx, plogs)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *stanzaToOtelConsumer) Logger() *zap.Logger {
+	return c.logger
+}
diff --git a/processor/signozlogspipelineprocessor/testdata/config.yaml b/processor/signozlogspipelineprocessor/testdata/config.yaml
new file mode 100644
index 00000000..52f84a4c
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/testdata/config.yaml
@@ -0,0 +1,8 @@
+operators:
+  - type: regex_parser
+    regex: '^(?P<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$'
+    timestamp:
+      parse_from: attributes.time
+      layout: '%Y-%m-%d %H:%M:%S'
+    severity:
+      parse_from: attributes.sev
diff --git a/processor/signozlogspipelineprocessor/utils.go b/processor/signozlogspipelineprocessor/utils.go
new file mode 100644
index 00000000..584ee418
--- /dev/null
+++ b/processor/signozlogspipelineprocessor/utils.go
@@ -0,0 +1,205 @@
+// Brought in as is from pkg/stanza/adapter/converter.go in opentelemetry-collector-contrib
+package signozlogspipelineprocessor
+
+import (
+	"fmt"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+func convertEntriesToPlogs(entries []*entry.Entry) plog.Logs {
+	resourceHashToIdx := make(map[uint64]int)
+	scopeIdxByResource := make(map[uint64]map[string]int)
+
+	pLogs := plog.NewLogs()
+	var sl plog.ScopeLogs
+
+	for _, e := range entries {
+		resourceID := adapter.HashResource(e.Resource)
+		var rl plog.ResourceLogs
+
+		resourceIdx, ok := resourceHashToIdx[resourceID]
+		if !ok {
+			resourceHashToIdx[resourceID] = pLogs.ResourceLogs().Len()
+
+			rl = pLogs.ResourceLogs().AppendEmpty()
+			upsertToMap(e.Resource, rl.Resource().Attributes())
+
+			scopeIdxByResource[resourceID] = map[string]int{e.ScopeName: 0}
+			sl = rl.ScopeLogs().AppendEmpty()
+			sl.Scope().SetName(e.ScopeName)
+		} else {
+			rl = pLogs.ResourceLogs().At(resourceIdx)
+			scopeIdxInResource, ok := scopeIdxByResource[resourceID][e.ScopeName]
+			if !ok {
+				scopeIdxByResource[resourceID][e.ScopeName] = rl.ScopeLogs().Len()
+				sl = rl.ScopeLogs().AppendEmpty()
+				sl.Scope().SetName(e.ScopeName)
+			} else {
+				sl = pLogs.ResourceLogs().At(resourceIdx).ScopeLogs().At(scopeIdxInResource)
+			}
+		}
+		convertInto(e, sl.LogRecords().AppendEmpty())
+	}
+
+	return pLogs
+}
+
+func upsertToMap(obsMap map[string]any, dest pcommon.Map) {
+	dest.EnsureCapacity(len(obsMap))
+	for k, v := range obsMap {
+		upsertToAttributeVal(v, dest.PutEmpty(k))
+	}
+}
+
+func upsertToAttributeVal(value any, dest pcommon.Value) {
+	switch t := value.(type) {
+	case bool:
+		dest.SetBool(t)
+	case string:
+		dest.SetStr(t)
+	case []string:
+		upsertStringsToSlice(t, dest.SetEmptySlice())
+	case []byte:
+		dest.SetEmptyBytes().FromRaw(t)
+	case int64:
+		dest.SetInt(t)
+	case int32:
+		dest.SetInt(int64(t))
+	case int16:
+		dest.SetInt(int64(t))
+	case int8:
+		dest.SetInt(int64(t))
+	case int:
+		dest.SetInt(int64(t))
+	case uint64:
+		dest.SetInt(int64(t))
+	case uint32:
+		dest.SetInt(int64(t))
+	case uint16:
+		dest.SetInt(int64(t))
+	case uint8:
+		dest.SetInt(int64(t))
+	case uint:
+		dest.SetInt(int64(t))
+	case float64:
+		dest.SetDouble(t)
+	case float32:
+		dest.SetDouble(float64(t))
+	case map[string]any:
+		upsertToMap(t, dest.SetEmptyMap())
+	case []any:
+		upsertToSlice(t, dest.SetEmptySlice())
+	default:
+		dest.SetStr(fmt.Sprintf("%v", t))
+	}
+}
+
+func upsertToSlice(obsArr []any, dest pcommon.Slice) {
+	dest.EnsureCapacity(len(obsArr))
+	for _, v := range obsArr {
+		upsertToAttributeVal(v, dest.AppendEmpty())
+	}
+}
+
+func upsertStringsToSlice(obsArr []string, dest pcommon.Slice) {
+	dest.EnsureCapacity(len(obsArr))
+	for _, v := range obsArr {
+		dest.AppendEmpty().SetStr(v)
+	}
+}
+
+func convertInto(ent *entry.Entry, dest plog.LogRecord) {
+	if !ent.Timestamp.IsZero() {
+		dest.SetTimestamp(pcommon.NewTimestampFromTime(ent.Timestamp))
+	}
+	dest.SetObservedTimestamp(pcommon.NewTimestampFromTime(ent.ObservedTimestamp))
+	dest.SetSeverityNumber(sevMap[ent.Severity])
+	if ent.SeverityText == "" {
+		dest.SetSeverityText(defaultSevTextMap[ent.Severity])
+	} else {
+		dest.SetSeverityText(ent.SeverityText)
+	}
+
+	upsertToMap(ent.Attributes, dest.Attributes())
+
+	if ent.Body != nil {
+		upsertToAttributeVal(ent.Body, dest.Body())
+	}
+
+	if ent.TraceID != nil {
+		var buffer [16]byte
+		copy(buffer[0:16], ent.TraceID)
+		dest.SetTraceID(buffer)
+	}
+	if ent.SpanID != nil {
+		var buffer [8]byte
+		copy(buffer[0:8], ent.SpanID)
+		dest.SetSpanID(buffer)
+	}
+	if ent.TraceFlags != nil && len(ent.TraceFlags) > 0 {
+		// The 8 least significant bits are the trace flags as defined in W3C Trace
+		// Context specification. Don't override the 24 reserved bits.
+		flags := uint32(ent.TraceFlags[0])
+		dest.SetFlags(plog.LogRecordFlags(flags))
+	}
+}
+
+var sevMap = map[entry.Severity]plog.SeverityNumber{
+	entry.Default: plog.SeverityNumberUnspecified,
+	entry.Trace:   plog.SeverityNumberTrace,
+	entry.Trace2:  plog.SeverityNumberTrace2,
+	entry.Trace3:  plog.SeverityNumberTrace3,
+	entry.Trace4:  plog.SeverityNumberTrace4,
+	entry.Debug:   plog.SeverityNumberDebug,
+	entry.Debug2:  plog.SeverityNumberDebug2,
+	entry.Debug3:  plog.SeverityNumberDebug3,
+	entry.Debug4:  plog.SeverityNumberDebug4,
+	entry.Info:    plog.SeverityNumberInfo,
+	entry.Info2:   plog.SeverityNumberInfo2,
+	entry.Info3:   plog.SeverityNumberInfo3,
+	entry.Info4:   plog.SeverityNumberInfo4,
+	entry.Warn:    plog.SeverityNumberWarn,
+	entry.Warn2:   plog.SeverityNumberWarn2,
+	entry.Warn3:   plog.SeverityNumberWarn3,
+	entry.Warn4:   plog.SeverityNumberWarn4,
+	entry.Error:   plog.SeverityNumberError,
+	entry.Error2:  plog.SeverityNumberError2,
+	entry.Error3:  plog.SeverityNumberError3,
+	entry.Error4:  plog.SeverityNumberError4,
+	entry.Fatal:   plog.SeverityNumberFatal,
+	entry.Fatal2:  plog.SeverityNumberFatal2,
+	entry.Fatal3:  plog.SeverityNumberFatal3,
+	entry.Fatal4:  plog.SeverityNumberFatal4,
+}
+
+var defaultSevTextMap = map[entry.Severity]string{
+	entry.Default: "",
+	entry.Trace:   "TRACE",
+	entry.Trace2:  "TRACE2",
+	entry.Trace3:  "TRACE3",
+	entry.Trace4:  "TRACE4",
+	entry.Debug:   "DEBUG",
+	entry.Debug2:  "DEBUG2",
+	entry.Debug3:  "DEBUG3",
+	entry.Debug4:  "DEBUG4",
+	entry.Info:    "INFO",
+	entry.Info2:   "INFO2",
+	entry.Info3:   "INFO3",
+	entry.Info4:   "INFO4",
+	entry.Warn:    "WARN",
+	entry.Warn2:   "WARN2",
+	entry.Warn3:   "WARN3",
+	entry.Warn4:   "WARN4",
+	entry.Error:   "ERROR",
+	entry.Error2:  "ERROR2",
+	entry.Error3:  "ERROR3",
+	entry.Error4:  "ERROR4",
+	entry.Fatal:   "FATAL",
+	entry.Fatal2:  "FATAL2",
+	entry.Fatal3:  "FATAL3",
+	entry.Fatal4:  "FATAL4",
+}